/*
 * Copyright (c) 2010-2025. Axon Framework
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.axonframework.serialization.avro;

import jakarta.annotation.Nonnull;
import jakarta.annotation.Nullable;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.message.SchemaStore;
import org.apache.avro.util.ClassUtils;
import org.axonframework.common.AxonConfigurationException;
import org.axonframework.serialization.AnnotationRevisionResolver;
import org.axonframework.serialization.ChainingContentTypeConverter;
import org.axonframework.serialization.Converter;
import org.axonframework.serialization.RevisionResolver;
import org.axonframework.serialization.SerializedObject;
import org.axonframework.serialization.SerializedType;
import org.axonframework.serialization.Serializer;
import org.axonframework.serialization.SimpleSerializedObject;
import org.axonframework.serialization.SimpleSerializedType;
import org.axonframework.serialization.UnknownSerializedType;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

import static org.axonframework.common.BuilderUtils.assertNonNull;
import static org.axonframework.common.BuilderUtils.assertThat;

/**
 * Serializer providing support for <a href="https://avro.apache.org/">Apache Avro</a>, using
 * <a href="https://avro.apache.org/docs/1.11.0/spec.html#single_object_encoding">Single Object Encoded binary
 * encoding</a>.
 * <p>
 * This serializer is intended to work for classes, representing messages specified by Avro Schema. It uses a delegate
 * serializer to serialize all other artifacts: metadata, tokens, sagas, snapshots and other non-message artifacts. In
 * general, it is a good idea to configure <code>event</code> and/or
 * <code>message</code> serializer to use <code>avro</code>, but let your <code>default</code> serializer
 * use <code>jackson</code> or another serializer of your choice.
 * </p>
 * <p>
 * The serialization/deserialization is delegated to the {@link AvroSerializerStrategy} implementations. By default, the
 * {@link SpecificRecordBaseSerializerStrategy} is provided and configured, to support Java classes generated by Avro
 * Maven plugin (delivered by Avro Java distribution). The {@link Builder} allows to register further strategies. At
 * least one strategy has to be configured for this serializer to work.
 * </p>
 *
 * @author Simon Zambrovski
 * @author Jan Galinski
 * @since 4.11.0
 * @deprecated In favor of Avro-specific {@link Converter} implementation.
 */
@Deprecated(forRemoval = true, since = "5.0.0")
public class AvroSerializer implements Serializer {

    private final RevisionResolver revisionResolver;
    private final Converter converter;
    private final List<AvroSerializerStrategy> serializerStrategies = new ArrayList<>();
    /*
     * Responsible for everything that is NOT Avro (e.g. MetaData, Tokens, Snapshots, Sagas).
     */
    private final Serializer serializerDelegate;

    /**
     * Creates the serializer instance.
     *
     * @param builder builder containing relevant settings.
     */
    protected AvroSerializer(@Nonnull Builder builder) {
        builder.validate();
        this.revisionResolver = builder.revisionResolver;
        this.serializerDelegate = builder.serializerDelegate;
        this.serializerStrategies.addAll(builder.serializerStrategies);
        this.converter = builder.converter;
        // converter registration
        if (this.converter instanceof ChainingContentTypeConverter chainingConverter) {
            chainingConverter.registerConverter(new ByteArrayToGenericRecordConverter(builder.schemaStore));
            Converter delegateConverter = serializerDelegate.getConverter();
            if (delegateConverter instanceof ChainingContentTypeConverter) {
                chainingConverter.setAdditionalConverters(((ChainingContentTypeConverter) delegateConverter).getContentTypeConverters());
            }
        }
    }

    /**
     * Creates a builder for Avro Serializer with defaults applied.
     * <p>
     * The {@link RevisionResolver} defaults to {@link AnnotationRevisionResolver}, the {@link Converter} defaults to
     * {@link ChainingContentTypeConverter} and the default serialization strategies are enabled, which effectively
     * activates the {@link SpecificRecordBaseSerializerStrategy}.
     * </p>
     * <p>
     * The following fields are mandatory, to create the Serializer:
     * <ul>
     *     <li>{@link Builder#schemaStore(SchemaStore)}, to lookup schemas by their fingerprints</li>
     *     <li>{@link Builder#serializerDelegate(Serializer)}, to deserialize all non-Avro artifacts</li>
     *     <li>at least one {@link AvroSerializerStrategy} either passed via
     *     {@link Builder#addSerializerStrategy(AvroSerializerStrategy)} or activated via
     *     {@link Builder#includeDefaultAvroSerializationStrategies(boolean)}, which is a default.</li>
     * </ul>
     * </p>
     *
     * @return fluent builder instance.
     */
    public static Builder builder() {
        return new Builder();
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> SerializedObject<T> serialize(Object object, @Nonnull Class<T> expectedRepresentation) {
        // assume: expectedRepresentation is byte[] (or possibly String), nothing else.
        // delegate null handling to serializerDelegate, since Avro can't serialize null.
        if (object != null) {
            Optional<AvroSerializerStrategy> serializerStrategy = serializerStrategies
                    .stream()
                    .filter(it -> it.test(object.getClass()))
                    .findFirst();

            if (serializerStrategy.isPresent()) {
                if (byte[].class.equals(expectedRepresentation)) {
                    return (SerializedObject<T>) serializerStrategy.get().serializeToSingleObjectEncoded(object);
                }
            }
        }

        // when we do not have an avro record, we delegate to jackson/xstream/...
        return serializerDelegate.serialize(object, expectedRepresentation);
    }

    @SuppressWarnings("unchecked")
    @Override
    public <S, T> T deserialize(@Nonnull SerializedObject<S> serializedObject) {
        if (SerializedType.isEmptyType(serializedObject.getType())) {
            return null;
        }
        Class<?> payloadType = classForType(serializedObject.getType());
        if (UnknownSerializedType.class.isAssignableFrom(payloadType)) {
            //noinspection unchecked
            return (T) new UnknownSerializedType(this, serializedObject);
        }

        Optional<AvroSerializerStrategy> serializerStrategy = serializerStrategies
                .stream()
                .filter(it -> it.test(payloadType))
                .findFirst();

        if (serializerStrategy.isPresent()) {

            // with upcasting:
            // GenericRecord -> T
            if (serializedObject.getContentType().equals(GenericRecord.class)) {
                return (T) serializerStrategy.get().deserializeFromGenericRecord(
                        (SerializedObject<GenericRecord>) serializedObject, payloadType
                );
            }

            // without upcasting:
            // byte[] -> T
            SerializedObject<byte[]> bytesSerialized;
            if (serializedObject.getContentType().equals(byte[].class)) {
                bytesSerialized = (SerializedObject<byte[]>) serializedObject;
            } else {
                bytesSerialized = new SimpleSerializedObject<>(convert(serializedObject.getData(), byte[].class),
                                                               byte[].class,
                                                               serializedObject.getType());
            }
            return (T) serializerStrategy.get().deserializeFromSingleObjectEncoded(bytesSerialized, payloadType);
        }

        // not an avro type, let delegate deal with it.
        return serializerDelegate.deserialize(serializedObject);
    }

    @Override
    public <T> boolean canSerializeTo(@Nonnull Class<T> expectedRepresentation) {
        return GenericRecord.class.equals(expectedRepresentation)
                || getConverter().canConvert(byte[].class, expectedRepresentation)
                || serializerDelegate.canSerializeTo(expectedRepresentation);
    }

    @Override
    @SuppressWarnings("rawtypes")
    public Class classForType(@Nonnull SerializedType type) {
        if (SimpleSerializedType.emptyType().equals(type)) {
            return Void.class;
        }
        try {
            return ClassUtils.forName(resolveClassName(type));
        } catch (ClassNotFoundException e) {
            return UnknownSerializedType.class;
        }
    }

    /**
     * Resolve the class name from the given {@code serializedType}. This method may be overridden to customize the
     * names used to denote certain classes, for example, by leaving out a certain base package for brevity.
     *
     * @param serializedType The serialized type to resolve the class name for
     * @return The fully qualified name of the class to load
     */
    protected String resolveClassName(SerializedType serializedType) {
        return serializedType.getName();
    }


    @SuppressWarnings("rawtypes")
    @Override
    public SerializedType typeForClass(@Nullable Class type) {
        if (type == null || Void.TYPE.equals(type) || Void.class.equals(type)) {
            return SimpleSerializedType.emptyType();
        }
        return new SimpleSerializedType(type.getName(), revisionResolver.revisionOf(type));
    }

    @Override
    public Converter getConverter() {
        return converter;
    }

    /**
     * Builder to set up Avro Serializer using some defaults. The {@link RevisionResolver} defaults to
     * {@link AnnotationRevisionResolver}, the {@link Converter} defaults to {@link ChainingContentTypeConverter} and
     * the default serialization strategies are enabled, which effectively activates the
     * {@link SpecificRecordBaseSerializerStrategy}.
     * <p>
     * The following fields are mandatory, to create the Serializer:
     * <ul>
     *     <li>{@link Builder#schemaStore(SchemaStore)}, to lookup schemas by their fingerprints</li>
     *     <li>{@link Builder#serializerDelegate(Serializer)}, to deserialize all non-Avro artifacts</li>
     *     <li>at least one {@link AvroSerializerStrategy} either passed via
     *     {@link Builder#addSerializerStrategy(AvroSerializerStrategy)} or activated via
     *     {@link Builder#includeDefaultAvroSerializationStrategies(boolean)}, which is a default.</li>
     * </ul>
     * </p>
     */
    public static class Builder {

        private final List<AvroSerializerStrategy> serializerStrategies = new ArrayList<>();
        private final AvroSerializerStrategyConfig.Builder configurationBuilder = AvroSerializerStrategyConfig
                .builder();
        private RevisionResolver revisionResolver = new AnnotationRevisionResolver();
        private SchemaStore schemaStore;
        private SchemaIncompatibilityChecker schemaIncompatibilityChecker = new DefaultSchemaIncompatibilityChecker();
        private Serializer serializerDelegate;
        private Converter converter = new ChainingContentTypeConverter();
        private boolean includeDefaultStrategies = true;

        /**
         * Sets revision resolver.
         *
         * @param revisionResolver revision resolver to use.
         * @return builder instance.
         */
        public Builder revisionResolver(@Nonnull RevisionResolver revisionResolver) {
            assertNonNull(revisionResolver, "RevisionResolver may not be null");
            this.revisionResolver = revisionResolver;
            return this;
        }

        /**
         * Sets schema store for Avro schema resolution.
         *
         * @param schemaStore schema store instance.
         * @return builder instance.
         */
        public Builder schemaStore(@Nonnull SchemaStore schemaStore) {
            assertNonNull(schemaStore, "SchemaStore may not be null");
            this.schemaStore = schemaStore;
            return this;
        }

        /**
         * Sets schema compatibility checker.
         */
        public Builder schemaIncompatibilityChecker(@Nonnull SchemaIncompatibilityChecker incompatibilityChecker) {
            assertNonNull(incompatibilityChecker, "SchemaIncompatibilityChecker may not be null");
            this.schemaIncompatibilityChecker = incompatibilityChecker;
            return this;
        }

        /**
         * Sets serializer delegate, used for all types which can't be converted to Avro.
         *
         * @param serializerDelegate serializer delegate.
         * @return builder instance.
         */
        public Builder serializerDelegate(@Nonnull Serializer serializerDelegate) {
            assertNonNull(serializerDelegate, "Serializer delegate may not be null");
            this.serializerDelegate = serializerDelegate;
            return this;
        }

        /**
         * Adds a serialization strategy.
         *
         * @param strategy strategy responsible for the serialization and deserialization.
         * @return builder instance.
         */
        public Builder addSerializerStrategy(@Nonnull AvroSerializerStrategy strategy) {
            assertNonNull(strategy, "AvroSerializerStrategy may not be null");
            this.serializerStrategies.add(strategy);
            return this;
        }

        /**
         * Sets the {@link Converter} used as a converter factory providing converter instances utilized by upcasters to
         * convert between different content types. Defaults to a {@link ChainingContentTypeConverter}.
         *
         * @param converter a {@link Converter} used as a converter factory providing converter instances utilized by
         *                  upcasters to convert between different content types
         * @return the current Builder instance, for fluent interfacing
         */
        public Builder converter(Converter converter) {
            assertNonNull(converter, "Converter may not be null");
            this.converter = converter;
            return this;
        }

        /**
         * Sets a flag controlling the instantiation of default {@link SpecificRecordBaseSerializerStrategy} and its
         * registration.
         * <p>
         * Allows to exclude the instantiation of the default strategy. If set to <code>false</code>, the caller must
         * provide at least one own strategy by calling {@link Builder#addSerializerStrategy(AvroSerializerStrategy)}
         * method.
         * </p>
         *
         * @param includeDefaultStrategies flag controlling default registration, defaults to <code>true</code>.
         * @return builder instance.
         */
        public Builder includeDefaultAvroSerializationStrategies(boolean includeDefaultStrategies) {
            this.includeDefaultStrategies = includeDefaultStrategies;
            return this;
        }

        /**
         * Sets a flag to perform Avro Schema Compatibility check during deserialization.
         *
         * @param performSchemaCompatibilityCheck flag to set, defaults to <code>true</code>.
         * @return builder instance.
         */
        public Builder performSchemaCompatibilityCheck(boolean performSchemaCompatibilityCheck) {
            this.configurationBuilder.performAvroCompatibilityCheck(performSchemaCompatibilityCheck);
            return this;
        }

        /**
         * Sets a flag to print relevant schemas in stack traces.
         *
         * @param includeSchemasInStackTraces flag to set, defaults to <code>false</code>.
         * @return builder instance.
         */
        public Builder includeSchemasInStackTraces(boolean includeSchemasInStackTraces) {
            this.configurationBuilder.includeSchemasInStackTraces(includeSchemasInStackTraces);
            return this;
        }

        /**
         * Validates whether the fields contained in this Builder are set accordingly.
         *
         * @throws AxonConfigurationException if one field is asserted to be incorrect according to the Builder's
         *                                    specifications
         */
        protected void validate() throws AxonConfigurationException {
            assertNonNull(revisionResolver, "RevisionResolver is mandatory");
            assertNonNull(schemaStore, "SchemaStore is mandatory");
            assertNonNull(serializerDelegate, "SerializerDelegate is mandatory");
            assertNonNull(schemaIncompatibilityChecker, "SchemaIncompatibilityChecker is mandatory");
            assertThat(serializerStrategies, (strategies) -> !strategies.isEmpty(),
                       "At least one AvroSerializerStrategy must be provided.");
        }

        /**
         * Creates an Avro Serializer instance.
         *
         * @return working instance.
         */
        public AvroSerializer build() {
            // include default strategy
            if (includeDefaultStrategies) {
                AvroSerializerStrategy defaultStrategy = new SpecificRecordBaseSerializerStrategy(
                        this.schemaStore,
                        this.revisionResolver,
                        this.schemaIncompatibilityChecker
                );
                this.addSerializerStrategy(defaultStrategy);
            }
            // create configuration
            AvroSerializerStrategyConfig configuration = configurationBuilder.build();
            // apply configuration to the strategies
            this.serializerStrategies.forEach(strategy -> strategy.applyConfig(configuration));
            return new AvroSerializer(this);
        }
    }
}
